package org.lemon.transport.rule.engine.mqtt.model;

import io.netty.channel.EventLoop;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.util.concurrent.Promise;
import org.lemon.transport.rule.engine.mqtt.handler.PendingOperation;
import org.lemon.transport.rule.engine.mqtt.handler.RetransmissionHandler;

import java.util.function.Consumer;

/**
 * 待发布的消息
 */
final public class MqttPendingPublish {

    private final int messageId;
    private final Promise<Void> future;
    private final MqttPublishMessage message;
    private final MqttQoS qos;

    private final RetransmissionHandler<MqttPublishMessage> publishRetransmissionHandler;
    private final RetransmissionHandler<MqttMessage> pubRelRetransmissionHandler;

    /**
     * 是否寄送
     */
    private boolean sent = false;

    public MqttPendingPublish(int messageId, Promise<Void> future, MqttPublishMessage message, MqttQoS qos,
                              PendingOperation operation) {
        this.messageId = messageId;
        this.future = future;
        this.message = message;
        this.qos = qos;

        this.publishRetransmissionHandler = new RetransmissionHandler<>(operation);
        this.publishRetransmissionHandler.setOriginalMessage(message);
        this.pubRelRetransmissionHandler = new RetransmissionHandler<>(operation);
    }

    public int getMessageId() {
        return messageId;
    }

    public Promise<Void> getFuture() {
        return future;
    }

    public boolean isSent() {
        return sent;
    }

    public void setSent(boolean sent) {
        this.sent = sent;
    }

    public MqttPublishMessage getMessage() {
        return message;
    }

    public MqttQoS getQos() {
        return qos;
    }

    public void startPublishRetransmissionTimer(EventLoop eventLoop, Consumer<Object> sendPacket) {
        this.publishRetransmissionHandler.setHandler(((fixedHeader, originalMessage) ->
                sendPacket.accept(new MqttPublishMessage(fixedHeader, originalMessage.variableHeader(), this.message.payload().retain()))));
        this.publishRetransmissionHandler.start(eventLoop);
    }

    public void onPubAckReceived() {
        this.publishRetransmissionHandler.stop();
    }

    public void setPubRelMessage(MqttMessage pubRelMessage) {
        this.pubRelRetransmissionHandler.setOriginalMessage(pubRelMessage);
    }

    public void startPubRelRetransmissionTimer(EventLoop eventLoop, Consumer<Object> sendPacket) {
        this.pubRelRetransmissionHandler.setHandler((fixedHeader, originalMessage) ->
                sendPacket.accept(new MqttMessage(fixedHeader, originalMessage.variableHeader())));
        this.pubRelRetransmissionHandler.start(eventLoop);
    }

    public void onPubCompReceived() {
        this.pubRelRetransmissionHandler.stop();
    }

    public void onChannelClosed() {
        this.publishRetransmissionHandler.stop();
        this.pubRelRetransmissionHandler.stop();
        if (this.message.payload() != null) {
            this.message.payload().release();
        }
    }
}
